package wangZhong.app.dws;
/**
 * @author wangz
 * @date 2022/9/2 18:03
 */

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.PropertyNamingStrategy;
import com.alibaba.fastjson.serializer.SerializeConfig;
import wangZhong.app.BaseAppV1;
import wangZhong.bean.TradeProvinceOrderWindow;
import wangZhong.common.Constant;
import wangZhong.util.AtguiguUtil;
import wangZhong.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;

public class Dws_DwsTradeProvinceOrderWindow extends BaseAppV1 {
    public static void main(String[] args) throws Exception {
        new Dws_DwsTradeProvinceOrderWindow().init(
                4010,
                2,
                "Dws_DwsTradeProvinceOrderWindow",
                Constant.TOPIC_DWD_TRADE_ORDER_DETAILS
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {

        //去重//
        SingleOutputStreamOperator<JSONObject> distinctedStream = distinctByOrderDetailId(stream);

        //把数据封装到pojo
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream = paseToPojo(distinctedStream);

        //聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> windowStream = windowAndAgg(beanStream);

        //写到doris
        writeTodoris(windowStream);

    }

    private void writeTodoris(SingleOutputStreamOperator<TradeProvinceOrderWindow> windowStream) {
        windowStream
                .map(bean ->{
                    SerializeConfig conf = new SerializeConfig();
                     conf.propertyNamingStrategy = PropertyNamingStrategy.SnakeCase;
                    String json = JSON.toJSONString(bean, conf);
                    System.out.println(json);
                    return JSON.toJSONString(bean,conf);
                })
                .addSink(FlinkSinkUtil.getDorisSink("edu.dws_trade_province_order_window"));
    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAndAgg(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
      return   beanStream
                .assignTimestampsAndWatermarks(
                        WatermarkStrategy
                        .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner((bean, ts) -> bean.getTs())
                )
                .keyBy(TradeProvinceOrderWindow::getProvinceId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .reduce(
                        new ReduceFunction<TradeProvinceOrderWindow>() {
                            @Override
                            public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow bean1,
                                                                   TradeProvinceOrderWindow bean2) throws Exception {

                                bean1.setOrderAmount(bean1.getOrderAmount().add(bean2.getOrderAmount()));
                                bean1.getOrderIdSet().addAll(bean2.getOrderIdSet());
                                bean1.getUserIdSet().addAll(bean2.getUserIdSet());
                                return bean1;
                            }
                        },
                        new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                            @Override
                            public void process(String courseId,
                                                Context context,
                                                Iterable<TradeProvinceOrderWindow> elements,
                                                Collector<TradeProvinceOrderWindow> collector) throws Exception {
                                TradeProvinceOrderWindow bean = elements.iterator().next();

                                bean.setStt(AtguiguUtil.toDateTime(context.window().getStart()));
                                bean.setEdt(AtguiguUtil.toDateTime(context.window().getEnd()));

                                bean.setCurDate(AtguiguUtil.toDate(System.currentTimeMillis()));

                                bean.setOrderCount((long) bean.getOrderIdSet().size());
                                bean.setUserCount((long) bean.getUserIdSet().size());

                                collector.collect(bean);

                            }
                        }
                );
    }

    private SingleOutputStreamOperator<TradeProvinceOrderWindow> paseToPojo(SingleOutputStreamOperator<JSONObject> distinctedStream) {
      return   distinctedStream
                .map(new MapFunction<JSONObject, TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow map(JSONObject object) throws Exception {
                       return TradeProvinceOrderWindow.builder()
                                .provinceId(object.getString(object.getString("province_id")))
                                .orderIdSet(new HashSet<>(Collections.singleton(object.getString("order_id"))))
                                .userIdSet(new HashSet<>(Collections.singleton(object.getString("user_id"))))
                                .orderAmount(object.getBigDecimal("final_amount"))
                                .ts(object.getLong("ts") * 1000)
                                .build();

                    }
                });

    }

    private SingleOutputStreamOperator<JSONObject> distinctByOrderDetailId(DataStreamSource<String> stream) {
      return   stream
                .map(JSON::parseObject)
                .keyBy(object -> object.getString("id"))
                .process(new KeyedProcessFunction<String, JSONObject, JSONObject>() {

                    private ValueState<JSONObject> maxtsdatestate;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        maxtsdatestate = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("maxTsDateState",JSONObject.class));
                    }

                    @Override
                    public void processElement(JSONObject object,
                                               Context context,
                                               Collector<JSONObject> collector) throws Exception {
                        if (maxtsdatestate.value() == null) {
                            maxtsdatestate.update(object);
                            context.timerService().registerProcessingTimeTimer(context.timerService().currentProcessingTime() + 5000L);
                        }else {
                            String last = maxtsdatestate.value().getString("row_op_ts");
                            String current = object.getString("row_op_ts");

                            if (AtguiguUtil.isLarger(current,last)) {
                                maxtsdatestate.update(object);
                            }
                        }
                    }

                    @Override
                    public void onTimer(long timestamp,
                                        OnTimerContext ctx,
                                        Collector<JSONObject> out) throws Exception {
                        out.collect(maxtsdatestate.value());
                    }
                });
    }
}
